/*
 * Copyright (C) 2007 James Deng (jtdeng@gmail.com)
 *
 * This program is free software; you can redistribute it and/or modify
 * it under the terms of the GNU General Public License as published by
 * the Free Software Foundation; either version 2 of the License, or
 * (at your option) any later version.
 *
 * This program is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 * GNU General Public License for more details.
 *
 * You should have received a copy of the GNU General Public License
 * along with this program; if not, write to the Free Software
 * Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
 *
 */


whenever sqlerror exit sql.sqlcode
prompt 'Creating package body plmsn_conversation...'

create or replace package body plmsn_conversation
as
	---------------------------------------------------------------
	-- Private Types
	---------------------------------------------------------------
	type conversation_rec_t is record (
		last_receive		pls_integer,
		last_idle			pls_integer,
		last_callee			varchar2(64), -- who is called to this conversation most recently
		ready				boolean,	--ready to send out messages
		conn2sb				utl_connection.connection
	);
	
	type conversation_aa_t is table of conversation_rec_t index by pls_integer;
	
	---------------------------------------------------------------
	-- Private member variables
	---------------------------------------------------------------
	m_http_proxy		varchar2(1024)	:= utl_config.get_cfg('http_proxy');
	--m_character_set		varchar2(32)	:= utl_config.get_cfg('character_set');
	m_passport			varchar2(64)	:= null;
	m_owner_id			pls_integer 	:= 0;
	m_conversation_list	conversation_aa_t;
	m_crlf				varchar2(10) := utl_connection.CRLF;

	---------------------------------------------------------------
	-- Private member procedures and functions
	---------------------------------------------------------------	
	function send_cmd_sync(conn IN OUT NOCOPY utl_connection.connection, cmd IN varchar2, args IN varchar2) return varchar2
	is
		l_cmd	varchar2(1024);
		l_resp	varchar2(32767);
		l_rc	pls_integer;
		l_trid	pls_integer;
	begin
		select SEQ_TRANSACTION_ID.nextval into l_trid from dual;
		l_cmd := cmd || ' ' || l_trid || ' ' || args;
		l_rc := utl_connection.write_line(conn, l_cmd);
		utl_logging.debug('package::plmsn_conversation.send_cmd_sync', 'SEND: ' || l_cmd);
		l_resp := utl_connection.get_line(conn, TRUE);		
		utl_logging.debug('package::plmsn_conversation.send_cmd_sync', 'RECV: ' || l_resp);		
		return l_resp;
	exception
		when others then
			utl_logging.error('package::plmsn_conversation.send_cmd_sync', '[COMMAND]: ' || l_cmd || m_crlf || SQLERRM || m_crlf || ' [TRACE]: ' || DBMS_UTILITY.format_error_backtrace);
	end;

	function send_cmd_async(conn IN OUT NOCOPY utl_connection.connection, cmd IN varchar2, args IN varchar2, payload IN RAW default null) return pls_integer
	is
		l_cmd	varchar2(32767);
		l_rc	pls_integer;
		l_trid	pls_integer; 
	begin
		select SEQ_TRANSACTION_ID.nextval into l_trid from dual;
		
		l_cmd := cmd || ' ' || l_trid || ' ' || args;
		l_rc := utl_connection.write_line(conn, l_cmd);

		if payload is not null 
		then
			l_rc := utl_connection.write_raw(conn, payload);
		end if;
		
		if payload is null 
		then
			utl_logging.debug('package::plmsn_conversation.send_cmd_async', 'SEND: ' || l_cmd );
		else	
			utl_logging.debug('package::plmsn_conversation.send_cmd_async', 'SEND: ' || l_cmd || utl_connection.CRLF || utl_raw.cast_to_varchar2(payload) );		
		end if;

		return l_trid;
	exception
		when others then
			utl_logging.warning('package::plmsn_conversation.send_cmd_async', '[COMMAND]: ' || l_cmd || m_crlf || SQLERRM || m_crlf || ' [TRACE]: ' || DBMS_UTILITY.format_error_backtrace);
	end;
		
	--send out all messages on this conversation channel
	procedure send_message(cvs_id IN pls_integer)
	is
		l_msg_raw raw(32767);
		l_trid pls_integer;
		l_msg_opt message_option%rowtype;
	begin	
		for msg_rec in (select *
						from outgoing_message_queue 
						where owner_id = m_owner_id and conversation_id = cvs_id order by msg_id asc)
		loop
			
			select * into l_msg_opt
			from message_option
			where option_name = msg_rec.msg_option;
						
			utl_mime.begin_mime;
			utl_mime.set_header('MIME-Version', '1.0');
			utl_mime.set_header('Content-Type', 'text/plain; charset=UTF-8');
			utl_mime.set_header('X-MMS-IM-Format', 'FN='|| l_msg_opt.FN ||'; EF='|| l_msg_opt.EF ||'; CO='|| l_msg_opt.CO ||'; CS='|| l_msg_opt.CS ||'; PF='|| l_msg_opt.PF );
			utl_mime.set_payload(msg_rec.msg_text);
			l_msg_raw := utl_raw.cast_to_raw(utl_mime.get_mime_text);
			l_trid := send_cmd_async(m_conversation_list(cvs_id).conn2sb, 'MSG', 'N ' || utl_raw.length(l_msg_raw), l_msg_raw );
			utl_mime.end_mime;

			update conversations c set c.status = 'BUSY' where c.conversation_id = cvs_id;			
			m_conversation_list(cvs_id).last_receive := dbms_utility.get_time();
			m_conversation_list(cvs_id).last_idle := dbms_utility.get_time();
			commit;
		end loop;

		delete outgoing_message_queue
		where owner_id = m_owner_id and conversation_id = cvs_id;
		commit;
	end;

	---------------------------------------------------------------
	-- MSN Conversation Handlers
	---------------------------------------------------------------	 	
	procedure handle_ANS(cvs_id IN pls_integer, args IN string_array)
	is
		l_result varchar2(8) := args(3);
	begin
		if l_result = 'OK' then
			update conversations c set c.status = 'BUSY' where c.conversation_id = cvs_id;
			commit;
			m_conversation_list(cvs_id).ready := true;
			m_conversation_list(cvs_id).last_receive := dbms_utility.get_time();
			m_conversation_list(cvs_id).last_idle := dbms_utility.get_time();
		end if;			
	end;
	
	procedure handle_USR(cvs_id IN pls_integer, args IN string_array)
	is
		l_result varchar2(8) := args(3);
	begin
		if l_result = 'OK' then
			update conversations c set c.status = 'BUSY' where c.conversation_id = cvs_id;			
			m_conversation_list(cvs_id).last_receive := dbms_utility.get_time();
			m_conversation_list(cvs_id).last_idle := dbms_utility.get_time();
			commit;
		end if;
	end;
	
	procedure handle_IRO(cvs_id IN pls_integer, args IN string_array)
	is
		l_attendees attendees_t := attendees_t();
	begin
		select attendees into l_attendees 
		from conversations
		where conversation_id = cvs_id;
		
		l_attendees.EXTEND(1);
		l_attendees( l_attendees.COUNT ) := args(5);
		
		update conversations
		set attendees = l_attendees
		where conversation_id = cvs_id;
		commit;
	end;
	
	procedure handle_MSG(cvs_id IN pls_integer, args IN string_array)
	is
		l_from		varchar2(64) := args(2);
		l_msg_len	pls_integer := args(4);
		l_msg_raw	RAW(32767);
		l_msg_text	varchar2(32767);
		l_content_type	varchar2(256);
		--l_echo_raw	RAW(32767);
	begin
		l_msg_raw := utl_connection.get_raw(m_conversation_list(cvs_id).conn2sb, l_msg_len);
		l_msg_text := utl_raw.cast_to_varchar2(l_msg_raw);

		update conversations c set c.status = 'BUSY' where c.conversation_id = cvs_id;
		m_conversation_list(cvs_id).last_receive := dbms_utility.get_time();
		m_conversation_list(cvs_id).last_idle := dbms_utility.get_time();

		utl_logging.debug('package::plmsn_conversation.handle_MSG', 'MESSAGE: ' || l_msg_text);
		
		utl_mime.begin_mime(l_msg_text);
		l_content_type := utl_mime.get_header('Content-Type');
		
		if instr(l_content_type, 'text/plain') > 0 then
			--When receives the text message
			insert into messages ( owner_id, conversation_id, direction, passport, msg_text )
			values (m_owner_id, cvs_id, 'FROM', l_from, utl_mime.get_payload );
			plmsn_events.OnMessageArrival(utl_mime.get_payload, l_from, cvs_id);
		elsif instr(l_content_type, 'text/x-msmsgscontrol') > 0 then
			--When someone is typing...
			null;
		elsif instr(l_content_type, 'text/x-weather') > 0 then
			--When receives the weather report message
			null;
		end if;
		
		utl_mime.end_mime;
		commit;		
	end;
	
	procedure handle_NAK(cvs_id IN pls_integer, args IN string_array)
	is
	begin
		m_conversation_list(cvs_id).ready := false;
	end;
	
	procedure handle_ACK(cvs_id IN pls_integer, args IN string_array)
	is
	begin
		null;
	end;

	procedure handle_CAL(cvs_id IN pls_integer, args IN string_array)
	is
	begin
		null;
	end;

	procedure handle_JOI(cvs_id IN pls_integer, args IN string_array)
	is
		l_attendees attendees_t := attendees_t();
	begin
		select attendees into l_attendees 
		from conversations
		where conversation_id = cvs_id;
		
		l_attendees.EXTEND(1);
		l_attendees( l_attendees.COUNT ) := args(2);
		
		update conversations
		set attendees = l_attendees
		where conversation_id = cvs_id;
		
		m_conversation_list(cvs_id).ready := true;
		update conversations c set c.status = 'BUSY' where c.conversation_id = cvs_id;
		m_conversation_list(cvs_id).last_receive := dbms_utility.get_time();
		m_conversation_list(cvs_id).last_idle := dbms_utility.get_time();
		
		commit;
	end;

	procedure handle_BYE(cvs_id IN pls_integer, args IN string_array)
	is
		l_who_left			varchar2(64) := args(2);
		l_all_attendees		attendees_t := attendees_t();
		l_rest_attendees	attendees_t := attendees_t();
	begin
		select attendees into l_all_attendees
		from conversations
		where conversation_id = cvs_id;
		
		select column_value
		bulk collect into l_rest_attendees
		from table(l_all_attendees)
		where column_value <> l_who_left;
		
		update conversations
		set attendees = l_rest_attendees
		where conversation_id = cvs_id;
		
		--If no attendees left, close this conversation
		if l_rest_attendees.COUNT() = 0 then
			update conversations set status = 'CLOSED' where conversation_id = cvs_id;
		end if;
		commit;
	end;
	
	procedure handle_ERROR(cvs_id IN pls_integer, args IN string_array)
	is
		l_error_code pls_integer := args(1);
	begin
		case l_error_code
			when 217 then	-- Called the principal, but it's not online
				update contacts set status = 'FLN' where owner_id = m_owner_id and passport = m_conversation_list(cvs_id).last_callee;
				update conversations set status = 'CLOSED' where conversation_id = cvs_id;
				commit;
				plmsn_services.stop_service(m_owner_id, m_conversation_list(cvs_id).last_callee);
			else
				null;
		end case;
	end;
	---------------------------------------------------------------
	-- MSN Conversation Dispatching
	---------------------------------------------------------------	 
	procedure dispatch_conversation( cvs_id IN pls_integer )
	is
		pragma autonomous_transaction;
		l_conversation	conversations%ROWTYPE;
		l_trid			pls_integer;
		l_sb_host		varchar2(32);
		l_sb_port		pls_integer;
		l_buffer		varchar2(32767);
		l_resp			varchar2(1024);
		args			string_array := string_array();
	begin
		select c.* into l_conversation from conversations c 
		where c.conversation_id = cvs_id;
				
		--ensure connection is established or destroyed before reading or writing anything
		case l_conversation.status
			when 'ISSUED' then
				l_sb_host := utl_string.split(l_conversation.switchboard, ':')(1);
				l_sb_port := utl_string.split(l_conversation.switchboard, ':')(2);
				m_conversation_list(cvs_id).conn2sb := utl_connection.open_connection(l_sb_host, l_sb_port, m_http_proxy);
				m_conversation_list(cvs_id).last_callee := null;
				
				--I was invited to the conversation
				if l_conversation.inviter <> m_passport then
					l_trid := send_cmd_async(m_conversation_list(cvs_id).conn2sb, 'ANS', m_passport || ' ' || l_conversation.authentication || ' ' || l_conversation.session_id);	
				else --I hosted the conversation
					l_resp := send_cmd_sync(m_conversation_list(cvs_id).conn2sb, 'USR', m_passport || ' ' || l_conversation.authentication);
					if regexp_instr(l_resp, 'USR [0-9]+ OK ' || m_passport) <> 0 
					then
						l_trid := send_cmd_async(m_conversation_list(cvs_id).conn2sb, 'CAL', l_conversation.initial_attendee);
						m_conversation_list(cvs_id).last_callee := l_conversation.initial_attendee;
					else
						update conversations set status = 'CLOSED' where conversation_id = cvs_id;
						commit;
						utl_logging.debug('package::plmsn_conversation.dispatch_conversation', 'Authentication to conversation failed');
						return;
					end if;	
				end if;
				
				update conversations set status = 'BUSY' where conversation_id = cvs_id;
				commit;
				m_conversation_list(cvs_id).last_receive := dbms_utility.get_time();
				m_conversation_list(cvs_id).last_idle := dbms_utility.get_time();
			when 'BUSY' then
				if (dbms_utility.get_time() - m_conversation_list(cvs_id).last_receive) > 10*100
				then
					update conversations set status = 'IDLE' where conversation_id = cvs_id;
					commit;
					m_conversation_list(cvs_id).last_idle := dbms_utility.get_time();
				end if;	
			when 'IDLE' then
				if (dbms_utility.get_time() - m_conversation_list(cvs_id).last_idle) > 60*100
				then
					update conversations set status = 'CLOSED' where conversation_id = cvs_id;
					commit;
					utl_logging.debug('package::plmsn_conversation.dispatch_conversation', 'IDLE TIMEDOUT');
					return;
				end if;	
			when 'CLOSED' then
				utl_connection.close_connection( m_conversation_list(cvs_id).conn2sb );
				m_conversation_list.delete(cvs_id);
				delete conversations where conversation_id = cvs_id;
				--if there are some messages still waiting on this conversation channel, remove these failed ones
				delete outgoing_message_queue where conversation_id = cvs_id;
				commit;
				return;
			else
				null;	
		end case;
				
		--reading incomings
		if utl_connection.available(m_conversation_list(cvs_id).conn2sb, 0) > 0
		then
			begin
				l_buffer := utl_connection.get_line(m_conversation_list(cvs_id).conn2sb, true);
				utl_logging.debug('package::plmsn_conversation.dispatch_conversation', 'RECV: ' || l_buffer);
				if l_buffer is not null
				then
					args := utl_string.split(l_buffer);
					case args(1)
						when 'ANS' then
							handle_ANS(cvs_id, args);
						when 'USR' then
							handle_USR(cvs_id, args);
						when 'IRO' then
							handle_IRO(cvs_id, args);
						when 'MSG' then
							handle_MSG(cvs_id, args);
						when 'NAK' then
							handle_NAK(cvs_id, args);
						when 'ACK' then
							handle_ACK(cvs_id, args);
						when 'CAL' then
							handle_CAL(cvs_id, args);
						when 'JOI' then
							handle_JOI(cvs_id, args);
						when 'BYE' then
							handle_BYE(cvs_id, args);
						else
							if regexp_instr(args(1), '[0-9]+') <> 0 then							
								handle_ERROR(cvs_id, args);
							end if;
					end case;
					args.DELETE;	
				end if;	
			exception
				when utl_connection.END_OF_INPUT or utl_connection.NETWORK_ERROR then
					update conversations set status = 'CLOSED' where conversation_id = cvs_id;
					m_conversation_list(cvs_id).ready := false;					
					commit;
					utl_logging.error('package::plmsn_conversation.dispatch_conversation', 'Lost connection to switchboard server: ' || m_crlf || SQLERRM || m_crlf || DBMS_UTILITY.format_error_backtrace);
			end;	
		end if;
		
		--writing outgoings if it's ready
		if m_conversation_list(cvs_id).ready then
			send_message(cvs_id);
		end if;
	end;
	
	
	---------------------------------------------------------------
	-- Public procedures
	---------------------------------------------------------------
	procedure set_owner_id(owner_id IN pls_integer)
	is
	begin
		m_owner_id := owner_id;
		select o.passport into m_passport from owners o where o.owner_id = m_owner_id;
	end;
	
	/*The main processing engine for switchboard sessions*/	
	procedure process
	is
	begin
		--Process all conversations of this owner
		for cvs_rec in (select conversation_id from conversations where owner_id = m_owner_id order by conversation_id asc)
		loop
			if not m_conversation_list.exists(cvs_rec.conversation_id)
			then
				m_conversation_list(cvs_rec.conversation_id).last_receive	:= 0;
				m_conversation_list(cvs_rec.conversation_id).last_idle		:= 0;
				m_conversation_list(cvs_rec.conversation_id).ready			:= false;
				m_conversation_list(cvs_rec.conversation_id).conn2sb		:= null;
			end if;
			
			dispatch_conversation(cvs_rec.conversation_id);
		end loop;
	end;

	/*Cleanup all dirty conversations, outgoing messages and service instances*/
	procedure cleanup
	is
		--pragma autonomous_transaction;
	begin
		--execute immediate 'truncate table conversations';
		--execute immediate 'truncate table outgoing_message_queue';
		delete conversations where owner_id = m_owner_id;
		delete outgoing_message_queue where owner_id = m_owner_id;
		
		--drop all existing scheduler jobs of this owner
		for svc_inst_rec in (	select t1.job_name 
								from service_instances t1, user_scheduler_jobs t2 
								where t1.owner_id = m_owner_id and t1.job_name = t2.job_name )
		loop
			DBMS_SCHEDULER.disable(svc_inst_rec.job_name, force => true);				
			DBMS_SCHEDULER.drop_job(svc_inst_rec.job_name, force => true);
		end loop;
		
		--execute immediate 'truncate table service_instances';
		delete service_instances where owner_id = m_owner_id;
		commit;
	exception
		when others then
			null;
	end;
	
	---------------------------------------------------------------
	-- Package Initializing
	---------------------------------------------------------------	 
--begin
--	cleanup;
end;
/
show err;
